package com.myflink.day04;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author Shelly An
 * @create 2020/9/19 11:23
 * flink 事件驱动（被动）  有数据才计算
 * spark 主动 到时间就计算
 * kafka 是事件驱动
 */
public class Window_TimeWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        DataStreamSource<String> socketDS = env.socketTextStream("122.51.172.104", 4444);

        //DataStream可以直接调用开窗的方法，但是都带“all”，这种情况下所有数据不分组，都在窗口里。开发中不建议使用
        //一般先分组，再使用开窗方法
        KeyedStream<Tuple2<String, Integer>, String> dataKS = socketDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        }).keyBy(data -> data.f0);

        //一个参数是滚动  两个参数是滑动
        //Time是Flink封装下的
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> data1WS = dataKS
                //滚动窗口  每5秒钟....
                .timeWindow(Time.seconds(5));

        //此处的sum是窗口的方法sum
        data1WS.sum(1).print("w1");

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> data2WS = dataKS
                //滑动窗口  每3秒钟统计一次 每5秒钟....
                .timeWindow(Time.seconds(5), Time.seconds(3));
        data2WS.sum(1).print("w2");

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> data3WS = dataKS
                //会话窗口 3秒钟还没来，就关闭窗口  withGap 间隔  每次会话统计一次
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(3)));
        data3WS.sum(1).print("w3");


        //和data1WS一个效果
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> data4WS = dataKS
                //底层
                .window(TumblingEventTimeWindows.of(Time.seconds(5)));

        //和data2WS一个效果
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> data5WS = dataKS
                //底层
                .window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(3)));




        env.execute();
    }
}
